/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.rocketmq.client.consumer.rebalance;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import org.apache.rocketmq.client.consumer.AllocateMessageQueueStrategy;
import org.apache.rocketmq.client.log.ClientLogger;
import org.apache.rocketmq.common.consistenthash.ConsistentHashRouter;
import org.apache.rocketmq.common.consistenthash.HashFunction;
import org.apache.rocketmq.common.consistenthash.Node;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.common.message.MessageQueue;

/**
 * Consistent Hashing queue algorithm
 * 一致性哈希策略
 * 一致性哈希有一个哈希环的概念，哈希环由数值 0到2^32-1 组成，不管内容多长的字符，经过哈希计算都能得到一个等长的数字，
 * 最后都会落在哈希环上的某个点，哈希环上的点都是虚拟的，比如我们这里使用Consumer的Id来进行哈希计算，得到的这几个是物理的点，
 * 然后把得到的点存到TreeMap里面，然后将所有的MessageQueue依次进行同样的哈希计算，得到距离MessageQueue顺时针方向最近的那个Consumer点，
 * 这个就是MessageQeueu最终归属的那个Consumer。
 */
public class AllocateMessageQueueConsistentHash implements AllocateMessageQueueStrategy {
    private final InternalLogger log = ClientLogger.getLog();

   /*
    一致性hash算法
    核心点：
    1. 哈希环。刚才的hash算法是对服务器取模，一致性哈希算法使用的是对2^32取模，
    就是一致性哈希将整个hash空间组织成了一个圆环，0-2^32-1.
    2.物理节点：将服务器（ip+端口）进行hash，映射成环上的一个节点。当请求到来时，根据请求的key，hash映射到环上，顺时针选取最近的一个服务器进行请求。
    3.虚拟节点：当环上的服务器较少的时候，会出现分配不均匀的情况，即大量的请求落在同一台服务器上。为了避免这种情况，就引入了虚拟节点，
    比如通过添加后缀的方式给物理节点克隆出三个虚拟节点，如果两台物理节点，都克隆三个虚拟节点，那么环上就一共有8个节点。
    只是被克隆的虚拟节点最后还是会定位到实际物理节点上，但是可以有效的分摊请求。
    一致性哈希相对于普通hash，优点在于映射到环上的其请求，是发送到环上离他最近的一个服务器，如果我们一台服务器宕机或者新增一台服务器，
    那么影响的请求只有这台服务器和前一个服务器节点之间的请求，其他的并不会影响。
    */

    /**
     * vNodeCount表示虚拟节点个数，默认10个
     */
    private final int virtualNodeCnt;

    /**
     * HashFunction表示哈希算法接口，可以自己实现，默认使用MD5实现哈希算法。
     */
    private final HashFunction customHashFunction;

    public AllocateMessageQueueConsistentHash() {
        this(10);
    }

    public AllocateMessageQueueConsistentHash(int virtualNodeCnt) {
        this(virtualNodeCnt, null);
    }

    public AllocateMessageQueueConsistentHash(int virtualNodeCnt, HashFunction customHashFunction) {
        if (virtualNodeCnt < 0) {
            throw new IllegalArgumentException("illegal virtualNodeCnt :" + virtualNodeCnt);
        }
        this.virtualNodeCnt = virtualNodeCnt;
        this.customHashFunction = customHashFunction;
    }

    /**
     * 一致性hash策略
     * 一致性哈希负载均衡的目的是要保证：相同的请求尽可能落在同一个服务器上。
     * 为什么是说尽可能？因为服务器会发生上下线，在少数服务器变化的时候不应该影响大多数的请求。
     * @param consumerGroup current consumer group
     * @param currentCID current consumer id
     * @param mqAll message queue set in current topic
     * @param cidAll consumer set in current consumer group
     * @return
     */
    @Override
    public List<MessageQueue> allocate(String consumerGroup, String currentCID, List<MessageQueue> mqAll,
        List<String> cidAll) {

        if (currentCID == null || currentCID.length() < 1) {
            throw new IllegalArgumentException("currentCID is empty");
        }
        if (mqAll == null || mqAll.isEmpty()) {
            throw new IllegalArgumentException("mqAll is null or mqAll empty");
        }
        if (cidAll == null || cidAll.isEmpty()) {
            throw new IllegalArgumentException("cidAll is null or cidAll empty");
        }

        List<MessageQueue> result = new ArrayList<MessageQueue>();
        // 如果这个消费者列表中不含包当前这个id, 就返回空，因为这个broker可能还不知道这个客户端的存在，就不用他干活了
        if (!cidAll.contains(currentCID)) {
            log.info("[BUG] ConsumerGroup: {} The consumerId: {} not in cidAll: {}",
                consumerGroup,
                currentCID,
                cidAll);
            return result;
        }
        // 将所有consumer变成节点 到时候经过hash计算 分布在hash环上
        Collection<ClientNode> cidNodes = new ArrayList<ClientNode>();
        for (String cid : cidAll) {
            cidNodes.add(new ClientNode(cid));
        }
        // 构建哈希环
        final ConsistentHashRouter<ClientNode> router; //for building hash ring
        if (customHashFunction != null) {
            router = new ConsistentHashRouter<ClientNode>(cidNodes, virtualNodeCnt, customHashFunction);
        } else {
            // 默认使用MD5进行Hash计算
            router = new ConsistentHashRouter<ClientNode>(cidNodes, virtualNodeCnt);
        }

        List<MessageQueue> results = new ArrayList<MessageQueue>();
        for (MessageQueue mq : mqAll) {
            // 对messageQueue进行hash计算，找到顺时针最近的consumer节点
            // 接着循环所有mq，通过router.routeNode方法找到mq映射的物理节点。
            ClientNode clientNode = router.routeNode(mq.toString());
            // 判断是否是当前consumer, 拿到物理节点后，和当前请求客户端id进行比较，是同一个，就把mq分配给他
            if (clientNode != null && currentCID.equals(clientNode.getKey())) {
                results.add(mq);
            }
        }

        return results;
    }

    @Override
    public String getName() {
        return "CONSISTENT_HASH";
    }

    private static class ClientNode implements Node {
        private final String clientID;

        public ClientNode(String clientID) {
            this.clientID = clientID;
        }

        @Override
        public String getKey() {
            return clientID;
        }
    }

}
